package njuflink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.flink.api.common.time.Time;
import java.sql.*;
import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.TimeUnit;



public class KafkaFlinkMySQL {
    public static void main(String[] args) throws Exception{
        //获取flink运行环境
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();

        //Kafka配置参数
        Properties properties=new Properties();
        properties.setProperty("bootstrap.servers",GlobalConfig.BOOTSRAP_SEVERS);
        properties.setProperty("group.id","languageclass");
        //指定key value的序列化方式
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", StringSerializer.class.getName());
        //指定topic名称
        String topic = GlobalConfig.TOPIC;
        //Kafka消费者
        FlinkKafkaConsumer<String> myConsumer=new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),properties);
        DataStream<String>stream=env.addSource(myConsumer);
        System.out.println("connect kafka");
        System.out.println(myConsumer);

        //对数据进行过滤
//        DataStream<String>filter=stream.filter((value) -> value.split(",").length==6);
//        DataStream<Tuple2<String, Integer> >newsCounts=filter.flatMap(new LineSplitter()).keyBy(0).sum(1);
        //stream=stream.filter((value)->value!=null);
//        DataStream<HashMap<String, Integer> >langnewsCounts=stream.map(new LanguageCount());
        //DataStream<String >langnewsCounts=stream.map(new LanguageCount1());
        DataStream<Tuple2<String,Long>>langnewsCounts1=stream.flatMap((String line, Collector<Tuple2<String,Long>> out)->{
            out.collect(Tuple2.of(line,1L));
        })
                .returns(Types.TUPLE(Types.STRING,Types.LONG));
        System.out.println("map");
        init_tb();
//        //自定义MySQLSink
//        newsCounts.addSink(new MySQLSink());
//        DataStream<Tuple2<String, Integer> >periodCounts=filter.flatMap(new LineSplitter2()).keyBy(0).sum(1);

//        langnewsCounts.addSink(new MySQLSink());
        KeyedStream<Tuple2<String,Long>,String> langnewCountsStream=langnewsCounts1.keyBy(data->data.f0);
        SingleOutputStreamOperator<Tuple2<String, Long>> languageSum = langnewCountsStream.sum(1);

        System.out.println(languageSum);
        languageSum.addSink(new MySQLSink2());

//        //自定义MySQLSink
//        periodCounts.addSink(new MySQLSink2());

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 尝试重启的次数
                Time.of(10, TimeUnit.SECONDS) // 间隔
        ));
        //执行Flink程序
        env.execute();

    }
    //public static Connection connection;
    public static void init_tb() throws ClassNotFoundException, SQLException {
        Connection connection;
        PreparedStatement preparedStatement;
        String tb=GlobalConfig.TABLE;
        //        //加载JDBC驱动
        Class.forName(GlobalConfig.DRIVER_CLASS);
        //获取数据库链接
        connection= DriverManager.getConnection(GlobalConfig.DB_URL,GlobalConfig.USER_NAME,GlobalConfig.PASSWORD);
        // 关闭自定提交
        connection.setAutoCommit(false);
        preparedStatement=connection.prepareStatement("SHOW TABLES LIKE \""+tb+"\"");
        ResultSet rs=preparedStatement.executeQuery();
        if(rs!=null){
            System.out.println("表已存在");
            String droptbsql = "DROP TABLE "+tb;
            preparedStatement.executeUpdate(droptbsql);
            System.out.println("表已删除");
        }
        preparedStatement=connection.prepareStatement("CREATE TABLE "+tb+"(language VARCHAR(25), count INT(11)) " );
        preparedStatement.executeUpdate();
        System.out.println("表创建成功");
        connection.commit();
        if(preparedStatement!=null){
            preparedStatement.close();
        }
        if(connection!=null){
            connection.close();
        }

    }
    public static HashMap<String,Integer> lang2cnt=new HashMap<String, Integer>();
//    public static final class LanguageCount implements MapFunction<String,HashMap<String,Integer>> {
//
//
//        public HashMap<String,Integer> map(String language) throws Exception {
//            if(!lang2cnt.containsKey(language)){
//                lang2cnt.put(language,1);
//            }
//            else{
//                lang2cnt.put(language,1+lang2cnt.get(language));
//            }
//            return lang2cnt;
//        }
//
//    }

    public static final class LanguageCount1 implements MapFunction<String,String> {


        public String map(String language) throws Exception {
            if(language!=null){
                if(!lang2cnt.containsKey(language)){
                    lang2cnt.put(language,1);
                }
                else{
                    lang2cnt.put(language,1+lang2cnt.get(language));
                }
                return language;
            }

            return "unknown";
        }

    }
}
